package com.sjc.zeye.product.function;

import com.sjc.zeye.product.domain.ProductViewCount;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

public class TopHotProduct extends KeyedProcessFunction<Tuple, ProductViewCount,String> {

    private int topN ;

    public TopHotProduct(int topN) {
        this.topN = topN;
    }
    // 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算
    private ListState<ProductViewCount> productState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 状态的注册
        ListStateDescriptor<ProductViewCount> descriptor = new ListStateDescriptor<>("product-state", ProductViewCount.class);
        productState = getRuntimeContext().getListState(descriptor);
    }

    // 里面的逻辑就很简单，目的就是求TopN
    @Override
    public void processElement(ProductViewCount input, Context context, Collector<String> collector) throws Exception {
        // 每条数据都保存到状态中
        productState.add(input);
        // 注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收齐了属于windowEnd窗口的所有商品数据
        context.timerService().registerEventTimeTimer(input.getWindowEnd() + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        // 获取收到的所有商品点击量
        List<ProductViewCount> allProduct = new ArrayList<>();
        Iterator<ProductViewCount> iterator = productState.get().iterator();
        while(iterator.hasNext()){
            allProduct.add(iterator.next()) ;
        }
        // 提前清除状态中的数据，释放空间
        productState.clear();
        // 按照点击量从大到小排序
        allProduct.sort(new Comparator<ProductViewCount>() {
            @Override
            public int compare(ProductViewCount o1, ProductViewCount o2) {
                return (int) (o2.getCount() - o1.getCount());
            }
        });

        // 将排名信息格式化成 String, 便于打印
        StringBuilder sb = new StringBuilder();
        sb.append("时间: ").append(new Timestamp(timestamp-1)).append("\n");
        int min = Math.min(topN,allProduct.size());
        for(int i=0; i< min; i++){
            ProductViewCount productViewCount = allProduct.get(i);
            sb.append("No").append(i).append(":")
                    .append("  商品ID=").append(productViewCount.getProductId())
                    .append("  浏览量=").append(productViewCount.getCount())
                    .append("\n");
        }
        sb.append("============================");
        out.collect(sb.toString());
    }
}
